/*
 * cm_log.c
 *
 *  Created on: 2015年10月26日
 *      Author: chens
 */

#include <stdlib.h>
#include <stdio.h>
#include <pthread.h>
#include <time.h>
#include <assert.h>
#include <string.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <fcntl.h>
#include <errno.h>
#include <sys/time.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/stat.h> // setup时循环创建文件路径时引入

#include "sqlite3.h"
#include "cm_log.h"

static int g_max_log_count = MAX_LOG_COUNT;
static enum log_level g_stdout_log_level = DEFAULT_PRINT_LEVEL;
static enum log_level g_db_log_level = DEFAULT_LOG_LEVEL;

static const char * log_lv_str[] = { "[DEBUG]", "[INFO] ", "[WARN] ", "[ERROR]",
        "[FATAL]" };

static char g_log_file[1024] = DEFAULT_LOG_FILE;

sqlite3 * g_database = NULL;
pthread_mutex_t g_database_lock = PTHREAD_MUTEX_INITIALIZER;
#define LOCK_DB pthread_mutex_lock(&g_database_lock)
#define UNLOCK_DB pthread_mutex_unlock(&g_database_lock)

/*
 * @brief   设置日志输出模式
 * @param   print_level: 大于等于这个级别的日志则输出到stdout
 *          write_db_level: 大于等于这个级别的日志则写入数据库
 *          不调用则使用 DEFAULT_LOG_LEVEL 和 DEFAULT_PRINT_LEVEL
 */
void log_set_output_level(enum log_level print_level,
        enum log_level write_db_level)
{
    if (print_level >= LOG_DEBUG && print_level <= LOG_FATAL)
        g_stdout_log_level = print_level;

    if (write_db_level >= LOG_DEBUG && write_db_level <= LOG_FATAL)
        g_db_log_level = write_db_level;
}

/*
 * @brief   设置最大记录条数，不调用默认使用 MAX_LOG_COUNT 宏
 *          count应当大于（远大于）MAX_CACHE定义的值
 */
void log_set_max_count(unsigned int count)
{
    if (count > MAX_CACHE)
        g_max_log_count = count;
}

/*
 * @brief   创建多级目录,path是路径名，不含文件名
 */
static inline int create_multi_dir(const char *path)
{
    int i, len;

    len = strlen(path);
    char dir_path[4096];
    memset(dir_path, 0, sizeof(dir_path));
    strncpy(dir_path, path, len);

    for (i = 0; i < len; i++) {
        if (dir_path[i] == '/' && i > 0) {
            dir_path[i] = '\0';
            if (access(dir_path, F_OK) < 0) {
                if (mkdir(dir_path, 0755) < 0) {
                    return -1;
                }
            }
            dir_path[i] = '/';
        }
    }

    if (i == len) {
        if (access(dir_path, F_OK) < 0)
            if (mkdir(dir_path, 0755) < 0)
                return -1;
    }

    return 0;
}

static void _mkdir_and_file(const char * fpath)
{
    char path[4096];
    char * pos = NULL;

    if (access(fpath, F_OK) == 0) // 文件已经存在
        return;

    pos = strrchr(fpath, '/');
    if (pos == NULL)
        return;

    memset(path, 0, sizeof(path));
    memcpy(path, fpath, pos - fpath);
    if (0 != create_multi_dir(path))
        printf("Error: create multi path failed.\n");
}

static const char * _SQL_CREATE_TABLE = "CREATE TABLE IF NOT EXISTS log("
        "id INTEGER PRIMARY KEY AUTOINCREMENT,"
        "timestamp INTEGER NOT NULL,"
        "level INTEGER NOT NULL,"
        "content TEXT NOT NULL"
        ")";
static const char * _SQL_CREATE_INDEX =
        "CREATE INDEX IF NOT EXISTS log_timestamp ON log (timestamp)";

/*
 * @brief   创建日志模块需要的表
 */
static inline int _create_table(sqlite3 * pdb)
{
    int ret;
    char * perrmsg = NULL;

    ret = sqlite3_exec(pdb, _SQL_CREATE_TABLE, NULL, NULL, &perrmsg);

    if (perrmsg != NULL) {
        printf("Error: %s\n", perrmsg);
        sqlite3_free(perrmsg);
        return ret;
    }

    ret = sqlite3_exec(pdb, _SQL_CREATE_INDEX, NULL, NULL, &perrmsg);

    if (perrmsg != NULL) {
        printf("Error: %s\n", perrmsg);
        sqlite3_free(perrmsg);
    }

    return ret;
}

#ifdef LOG_WRITE_CODE
static void* _log_thread(void *);
#endif

/*
 * @brief   初始化日志模块，指定数据文件路径，启动缓存管理线程
 * @param   fpath: 日志文件(db)全路径
 * @warning 必须先于其他函数调用一次，否则无法正确记录日志
 */
void log_setup(const char *fpath)
{
    if (fpath != NULL) {
        memset(g_log_file, 0, sizeof(g_log_file));
        snprintf(g_log_file, sizeof(g_log_file), "%s", fpath);
    }

    _mkdir_and_file(g_log_file);

    LOCK_DB;
    if (g_database != NULL)
        sqlite3_close(g_database);

    if (SQLITE_OK != sqlite3_open(g_log_file, &g_database)) {
        printf("Error: open database(%s) failed.\n", g_log_file);
        g_database = NULL;
        UNLOCK_DB;
        return;
    }

    if (SQLITE_OK != _create_table(g_database)) {
        sqlite3_close(g_database);
        UNLOCK_DB;
        return;
    }
    UNLOCK_DB;

#ifdef LOG_WRITE_CODE
    static int thread_started = 0;
    if (thread_started)
        return;

    pthread_t tid;
    pthread_attr_t attr;
    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);

    if (0 == pthread_create(&tid, &attr, _log_thread, NULL))
        thread_started = 1;

    pthread_attr_destroy(&attr);
#endif
}

void log_destroy()
{
    pthread_mutex_destroy(&g_database_lock);
    if (g_database != NULL) {
        sqlite3_close(g_database);
        g_database = NULL;
    }
}

#ifdef LOG_WRITE_CODE

struct log_item {
    enum log_level level;
    time_t timestamp;
    char content[MAX_LOG_LEN];
};

/*
 * @brief   全局的缓冲区, 是一个数组，通过标记读写位置来实现缓冲区控制
 */
static struct {
    unsigned r_idx;
    unsigned w_idx;
    pthread_mutex_t lock;
    pthread_cond_t cond_full;
    struct log_item logs[MAX_CACHE];
} g_log_cache = { 0, 0, PTHREAD_MUTEX_INITIALIZER, PTHREAD_COND_INITIALIZER };

#define CACHE_FULL  (((g_log_cache.w_idx + 1) % MAX_CACHE) == g_log_cache.r_idx)
#define CACHE_EMPTY (g_log_cache.w_idx == g_log_cache.r_idx)

/*
 * @brief   返回cache中有多少空间可用，与 cache 的读写位置关系相关
 */
static inline int _cache_space()
{
    if (g_log_cache.r_idx == g_log_cache.w_idx) {
        return MAX_CACHE; // empty
    } else if (g_log_cache.r_idx > g_log_cache.w_idx) {
        return (g_log_cache.r_idx - g_log_cache.w_idx);
    } else {
        return (MAX_CACHE - (g_log_cache.w_idx - g_log_cache.r_idx));
    }
}

#define LOCK_CACHE    pthread_mutex_lock(&(g_log_cache.lock))
#define UNLOCK_CACHE    pthread_mutex_unlock(&(g_log_cache.lock))
#define FULL_WAKEUP    pthread_cond_signal(&(g_log_cache.cond_full))

#define TIMED_WAIT_FULL  do { \
    struct timespec t = {time(NULL) + 5, 0}; \
    pthread_mutex_lock(&(g_log_cache.lock));\
    pthread_cond_timedwait(&(g_log_cache.cond_full), &(g_log_cache.lock), &t);\
    pthread_mutex_unlock(&(g_log_cache.lock));\
} while(0);

/*
 * @brief   将日志加入缓存，当缓存满的时候则写入数据库
 * @warning 当缓冲区满时，可能需要等待一段时间才能返回
 */
static void _log_append(enum log_level lv, time_t timestamp,
        const char * content)
{
    struct log_item * ptr = NULL;
    int flush_flag = 0;

    while (1) {
        LOCK_CACHE;
        if (CACHE_FULL) {
            UNLOCK_CACHE;
//            FULL_WAKEUP;
            usleep(1000);
        } else {
            break;
        }
    }

    ptr = &(g_log_cache.logs[g_log_cache.w_idx]);
    ptr->level = lv;
    ptr->timestamp = timestamp;
    strncpy(ptr->content, content, MAX_LOG_LEN - 1);
    ptr->content[MAX_LOG_LEN - 1] = 0;

    g_log_cache.w_idx = (g_log_cache.w_idx + 1) % MAX_CACHE;

    //cache用掉一半时, 把日志全部写入数据库
    if (_cache_space() < MAX_CACHE / 2)
        flush_flag = 1;

    UNLOCK_CACHE;

    if (flush_flag) {
        FULL_WAKEUP;
    }
}

/*
 * @brief   把缓存全部插入数据库, 在_flush 里面调用
 */
static inline int _insert_cache(sqlite3 *pdb)
{
    int ret;
    char * perrmsg = NULL;
    char * sql = NULL;
    char temp[MAX_LOG_LEN + 128];
    int sql_len = 0;
    unsigned wpos = 0;
    unsigned rpos = 0;

    LOCK_CACHE;

    wpos = g_log_cache.w_idx;
    sql_len = MAX_LOG_LEN * (MAX_CACHE - _cache_space()) + 4096;

    sql = (char *) calloc(1, sql_len);
    if (sql == NULL) {
        printf("Error: malloc for sql failed.\n");
        UNLOCK_CACHE;
        return SQLITE_ERROR;
    }
    snprintf(temp, sizeof(temp), "%s",
            "INSERT INTO log (timestamp, level, content) VALUES ");
    strcat(sql, temp);

    rpos = g_log_cache.r_idx;
    for (; rpos != wpos; rpos = (rpos + 1) % MAX_CACHE) {
        if (rpos != g_log_cache.r_idx) // 不是第一个则加上分割用的逗号
            strcat(sql, ", ");
        struct log_item * ptr = &(g_log_cache.logs[rpos]);

        struct tm t = *localtime(&(ptr->timestamp));
        snprintf(temp, sizeof(temp),
                "(%ld, %d, \"%.4d-%.2d-%.2d %.2d:%.2d:%.2d %s  %s\")",
                ptr->timestamp, ptr->level, t.tm_year + 1900, t.tm_mon + 1,
                t.tm_mday, t.tm_hour, t.tm_min, t.tm_sec,
                log_lv_str[ptr->level], ptr->content);
        strcat(sql, temp);
    }

//    printf("final sql: %s\n", sql);
    ret = sqlite3_exec(pdb, sql, NULL, NULL, &perrmsg);
    if (perrmsg != NULL) {
        printf("Error: %s\n", perrmsg);
        sqlite3_free(perrmsg);
    }

    g_log_cache.r_idx = wpos; // 写完后移动r_idx, r_idx只在此修改因此不用锁

    free(sql);

    UNLOCK_CACHE;

    return ret;
}

static inline void _delete(sqlite3 *pdb)
{
    char * perrmsg = NULL;
    char sql[1024];

    assert(g_max_log_count > MAX_CACHE);

    LOCK_DB;

    snprintf(sql, 1024, "delete from log where id not in ("
            "select id from log "
            "order by id DESC limit %d"
            ")", g_max_log_count - MAX_CACHE);
    ;
    sqlite3_exec(pdb, sql, NULL, NULL, &perrmsg);
    if (perrmsg != NULL) {
        printf("Error: %s\n", perrmsg);
        sqlite3_free(perrmsg);
    }

    UNLOCK_DB;
}

/*
 * @brief   刷缓存，比较耗时，在 _log_thread 中调用
 */
static inline void _flush(sqlite3 *pdb)
{
    LOCK_CACHE;
    if (CACHE_EMPTY) {
        UNLOCK_CACHE;
        return; // 没有缓存则直接退出
    }
    UNLOCK_CACHE;

    LOCK_DB;
    _insert_cache(pdb);
    UNLOCK_DB;
}

/*
 * @brief   等待信号刷新cache，且等待超时为5秒, 保证每5秒刷新一次cache
 */
static void *_log_thread(void *arg)
{
    while (1) {
        TIMED_WAIT_FULL

        if (g_database == NULL) {
            printf("Error: database(%s) was not opened.\n", g_log_file);
            UNLOCK_DB;
            continue;
        }

        _delete(g_database); // 删除多余的记录
        _flush(g_database);

    }
    return NULL;
}

/************************ 以下是对外接口 ***********************/

/*
 * @brief   刷新日志模块的缓存，全部写入数据库
 * @warning 写数据库可能比较耗时(100ms级)，频繁调用可能降低程序性能
 */
void log_flush()
{
    if (g_database)
        _flush(g_database);
}

/*
 * @brief   记录日志接口
 */
void log_impl(enum log_level level, const char * content)
{
    if (content == NULL || content[0] == 0 || g_database == NULL)
        return;

    if (level >= g_stdout_log_level)
        printf("%s\n", content);

    if (level >= g_db_log_level)
        _log_append(level, time(NULL), content);

    return;
}

#endif // LOG_WRITE_CODE

#ifdef LOG_QUERY_CODE

int log_query_time(time_t begin, time_t end, int max, char **buf)
{
    return log_query(LOG_DEBUG, LOG_FATAL, begin, end, max, buf);
}

int log_query_level(enum log_level lv, time_t begin, time_t end, int max,
        char **buf)
{
    return log_query(lv, lv, begin, end, max, buf);
}

int log_query(enum log_level lv_begin, enum log_level lv_end, time_t begin,
        time_t end, int max, char **buf)
{
    return log_query_order(lv_begin, lv_end, begin, end, max, buf, LOG_ID_DESC);
}

static const char * G_SELECT_DATA_TIME_ASC = "select content from log "
        "where timestamp >= %d and timestamp <= %d and "
        "level >= %d and level <= %d "
        "order by timestamp ASC, id ASC limit %d";
static const char * G_SELECT_CNT_TIME_ASC =
        "select count(*) from (select * from log "
                "where timestamp >= %d and timestamp <= %d and "
                "level >= %d and level <= %d "
                "order by timestamp ASC limit %d)";

static const char * G_SELECT_DATA_TIME_DESC = "select content from log "
        "where timestamp >= %d and timestamp <= %d and "
        "level >= %d and level <= %d "
        "order by timestamp DESC, id DESC limit %d";
static const char * G_SELECT_CNT_TIME_DESC =
        "select count(*) from (select * from log "
                "where timestamp >= %d and timestamp <= %d and "
                "level >= %d and level <= %d "
                "order by timestamp DESC limit %d)";

static const char * G_SELECT_DATA_ID_ASC = "select content from log "
        "where timestamp >= %d and timestamp <= %d and "
        "level >= %d and level <= %d "
        "order by id ASC limit %d";
static const char * G_SELECT_CNT_ID_ASC =
        "select count(*) from (select * from log "
                "where timestamp >= %d and timestamp <= %d and "
                "level >= %d and level <= %d "
                "order by id ASC limit %d)";

static const char * G_SELECT_DATA_ID_DESC = "select content from log "
        "where timestamp >= %d and timestamp <= %d and "
        "level >= %d and level <= %d "
        "order by id DESC limit %d";
static const char * G_SELECT_CNT_ID_DESC =
        "select count(*) from (select * from log "
                "where timestamp >= %d and timestamp <= %d and "
                "level >= %d and level <= %d "
                "order by id DESC limit %d)";

int log_query_order(enum log_level lv_begin, enum log_level lv_end,
        time_t begin, time_t end, int max, char **buf, enum log_order order)
{
    const char * data_sql = NULL;
    const char * count_sql = NULL;
    int total_len = 0;
    int count = 0;
    int offset;
    char * temp;
    char sql[1024];
    sqlite3_stmt * pstmt;

    switch (order) {
    case LOG_TIME_ASC:
        data_sql = G_SELECT_DATA_TIME_ASC;
        count_sql = G_SELECT_CNT_TIME_ASC;
        break;
    case LOG_TIME_DESC:
        data_sql = G_SELECT_DATA_TIME_DESC;
        count_sql = G_SELECT_CNT_TIME_DESC;
        break;
    case LOG_ID_ASC:
        data_sql = G_SELECT_DATA_ID_ASC;
        count_sql = G_SELECT_CNT_ID_ASC;
        break;
    case LOG_ID_DESC:
        data_sql = G_SELECT_DATA_ID_DESC;
        count_sql = G_SELECT_CNT_ID_DESC;
        break;
    }

    *buf = NULL;

    if (g_database == NULL)
        return 0;

    snprintf(sql, 1024, count_sql, begin, end, lv_begin, lv_end, max);
    sqlite3_prepare_v2(g_database, sql, -1, &pstmt, NULL);
    if (SQLITE_ROW == sqlite3_step(pstmt)) {
        count = sqlite3_column_int(pstmt, 0);
        total_len = count * MAX_LOG_LEN;
        printf("count = %d, total len = %d\n", count, total_len);
    }
    sqlite3_finalize(pstmt);

    if (total_len == 0)
        return 0;

    *buf = (char *) calloc(1, total_len);
    if (*buf == NULL)
        return 0;

    snprintf(sql, 1024, data_sql, begin, end, lv_begin, lv_end, max);
    sqlite3_prepare_v2(g_database, sql, -1, &pstmt, NULL);
    offset = 0;
    temp = *buf;
    while (SQLITE_ROW == sqlite3_step(pstmt)) {
        char * content = (char *) sqlite3_column_text(pstmt, 0);
        snprintf(temp, total_len - offset - 1, "%s\n", content);

        offset += strlen(temp);
        temp = *buf + offset;
    }
    sqlite3_finalize(pstmt);

    return offset;
}

/*
 * @brief   查询当前状态
 * @return  在三个出参中分别返回
 *              count: 当前一共有多少条记录，
 *              begin: 最早一条记录的时间
 *              end: 最晚一条记录的时间
 */
int log_query_stat(int *count, time_t *begin, time_t *end)
{
    static const char * _SQL =
            "select count(*), min(timestamp), max(timestamp) from log";
    sqlite3_stmt * pstmt;

    *count = 0;
    *begin = 0;
    *end = 0;

    if (g_database == NULL)
        return 0;

    if (count == NULL || begin == NULL || end == NULL)
        return -1;

    sqlite3_prepare_v2(g_database, _SQL, -1, &pstmt, NULL);
    if (SQLITE_ROW == sqlite3_step(pstmt)) {
        *count = sqlite3_column_int(pstmt, 0);
        *begin = (time_t) sqlite3_column_int(pstmt, 1);
        *end = (time_t) sqlite3_column_int(pstmt, 2);
    }

    sqlite3_finalize(pstmt);

    return 0;
}

/*
 * @brief   释放查询接口所返回的buffer
 */
void log_freebuf(char *buf)
{
    if (buf != NULL)
        free(buf);
}

#endif

#if defined(LOG_CLIENT_CODE) || defined(LOG_SERVER_CODE)

static inline int _write_back(int fd, const char *msg, unsigned len)
{
    int ret;
    unsigned offset = 0;

    while (offset < len) {
        ret = write(fd, msg + offset, len - offset);
        if (ret < 0)
            log_error("_write_back failed: %s", strerror(errno));
        if (ret < 0)
            return -1;

        offset += ret;
    }

    return 0;
}

static inline int _read_n_bytes(int fd, char *buf, unsigned n)
{
    int ret;
    unsigned offset = 0;

    while (offset < n) {
        ret = read(fd, buf + offset, n - offset);
        if (ret < 0)
            log_error("_read_n_bytes failed: %s", strerror(errno));
        if (ret <= 0)
            return -1;
        offset += ret;
    }
    return 0;
}
#endif

#ifdef LOG_CLIENT_CODE

/*
 * @brief   创建日志客户端
 */
int log_client_create(const char * server)
{
    int fd;
    struct sockaddr_in addr;
    int flags;
    struct timeval timeout = { CREATE_LOG_CLIENT_TIMEOUT, 0 };
    fd_set rdset, wrset, eset;
    int ret;

    if (server == NULL)
        return -1;

    fd = socket(AF_INET, SOCK_STREAM, 0);
    if (-1 == fd)
        return -1;

    flags = fcntl(fd, F_GETFL, 0);
    fcntl(fd, F_SETFL, flags | O_NONBLOCK);

    addr.sin_family = AF_INET;
    addr.sin_port = htons(LOG_PORT);
    addr.sin_addr.s_addr = inet_addr(server);

    ret = connect(fd, (struct sockaddr *) &addr, sizeof(addr));
    if (ret != 0 && errno != EINPROGRESS) {
        perror("connect error: ");
        goto _L_ERROR;
    } else if (ret == 0) {
        goto _L_OK;
    }

    FD_ZERO(&rdset);
    FD_SET(fd, &rdset);
    wrset = rdset;
    eset = rdset;

    ret = select(fd + 1, &rdset, &wrset, &eset, &timeout);
    if (ret <= 0) {
        perror("select error.");
        goto _L_ERROR;
    }

    if (!FD_ISSET(fd, &rdset) && !FD_ISSET(fd, &wrset)) {
        printf("read/write not set.\n");
        goto _L_ERROR;
    }

    int err;
    socklen_t len = (socklen_t) sizeof(int);
    if (getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0) {
        perror("getsockopt: ");
        goto _L_ERROR;
    }

    if (err != 0) {
        printf("err == %d\n", err);
        goto _L_ERROR;
    }

    _L_OK: printf("Connect to server(%s) success.\n", server);
    fcntl(fd, F_SETFL, flags);
    return fd;

    _L_ERROR: printf("Connect to server(%s) failed.\n", server);
    close(fd);
    return -1;
}

/*
 * @brief   封装查询命令并发送给服务器
 */
static inline int _send_query_cmd(int fd, enum log_level lv_begin,
        enum log_level lv_end, time_t begin, time_t end, int max)
{
    char buffer[MAX_LOG_CMD_LEN];
    struct log_head *head;
    int len;

    memset(buffer, 0, sizeof(buffer));
    head = (struct log_head *) buffer;

    sprintf(head->version, "%s", LOG_VERSION);
    sprintf(head->command, "QUERY");
    sprintf(&(head->content_start), "%d %d %ld %ld %d", lv_begin, lv_end, begin,
            end, max);

    len = strlen(&(head->content_start));
    sprintf(head->length, "%d", len);

    printf("send query cmd: %s %s %s %s\n", head->version, head->command,
            head->length, (char *) (&head->content_start));
    return _write_back(fd, buffer, LOG_HEAD_SIZE + len);
}

/*
 * @brief   向服务器发送查询命令，并等待返回结果，查询结果保存在buf指向的指针处
 *          buf在函数内部分配，使用完后必须调用 log_client_freebuf 进行释放
 * @return  -2: 服务器端错误，错误信息在buf中
 *          -1: 表示fd错误，应当关闭fd
 *          0 : 没有查询结果
 *          >0: 表示buf中存放的数据大小
 */
int log_client_query(int fd, enum log_level lv_begin, enum log_level lv_end,
        time_t begin, time_t end, int max, char **buf)
{
    int len;
    int ret;
    struct log_head head;

    *buf = NULL;

    ret = _send_query_cmd(fd, lv_begin, lv_end, begin, end, max);
    if (ret != 0)
        return -1;

    memset(&head, 0, sizeof(head));

    len = read(fd, &head, LOG_HEAD_SIZE);
    if (len != LOG_HEAD_SIZE) {
        printf("Read header failed.\n");
        return -1;
    }

    len = atoi(head.length);
    printf("Result from server: %d\n", len);
    if (len <= 0)
        return 0;

    *buf = (char *) calloc(1, len + 1);
    if (*buf == NULL) {
        printf("malloc for query result failed.\n");
        return -1; // 分配失败
    }
    len = _read_n_bytes(fd, *buf, len);
    if (len != 0) {
        free(*buf);
        *buf = NULL;
        return -1;
    }
    if (strcmp(head.command, "RESULT") == 0) {
        return len;
    } else {
        return 0;
    }
}

static inline int _send_stat_cmd(int fd)
{
    struct log_head head;

    memset(&head, 0, sizeof(head));

    sprintf(head.version, "%s", LOG_VERSION);
    sprintf(head.command, "%s", "STAT_QUERY");
    sprintf(head.length, "%d", 0);

    printf("Send stat cmd: %s %s %s\n", head.version, head.command,
            head.length);
    return _write_back(fd, (char *) &head, LOG_HEAD_SIZE);
}

/*
 * @brief   查询当前状态
 * @return  在三个出参中分别返回
 *              count: 当前一共有多少条记录，
 *              begin: 最早一条记录的时间
 *              end: 最晚一条记录的时间
 */
int log_client_query_stat(int fd, int *count, time_t *begin, time_t *end)
{
    char temp[MAX_LOG_CMD_LEN];
    struct log_head head;
    int len;
    int ret;

    if (0 != _send_stat_cmd(fd)) {
        printf("Write to server failed.\n");
        return -1;
    }

    memset(&head, 0, sizeof(head));
    ret = read(fd, &head, LOG_HEAD_SIZE);
    if (ret != LOG_HEAD_SIZE) {
        printf("Read answer failed.\n");
        return -1;
    } else {
        printf("Answer %s %s %s\n", head.version, head.command, head.length);
    }

    len = atoi(head.length);
    if (len <= 0) {
        printf("Error len = 0\n");
        return 0;
    } else {
        printf("Answer len = %d\n", len);
    }

    memset(temp, 0, sizeof(temp));
    if (0 != _read_n_bytes(fd, temp, len)) {
        printf("Read (%d) for answer failed.\n", len);
        return -1;
    } else {
        printf("Answer from server : %s\n", temp);
    }

    if (strcmp("STATE", head.command) == 0) {
        if (3 != sscanf(temp, "%d %ld %ld", count, begin, end)) {
            printf("Answer format failed.\n");
            *count = 0;
            *begin = 0;
            *end = 0;
            return 0;
        } else {
            printf("Answer from server: count = %d, begin = %ld, end = %ld\n",
                    *count, *begin, *end);
            return 1;
        }
    } else {
        return 0;
    }
}

static inline int _send_write_cmd(int fd, enum log_level lv,
        const char * content)
{
    char buf[MAX_LOG_CMD_LEN + MAX_LOG_LEN];
    struct log_head * head;
    int len;

    memset(buf, 0, sizeof(buf));
    head = (struct log_head *) buf;

    sprintf(head->version, "%s", LOG_VERSION);
    sprintf(head->command, "LOG %d", lv);
    sprintf(&(head->content_start), "%s", content);
    len = strlen(&(head->content_start));
    sprintf(head->length, "%d", len);

    printf("send write cmd: %s %s %s %s\n", head->version, head->command,
            head->length, (char *) (&head->content_start));

    return _write_back(fd, buf, LOG_HEAD_SIZE + len);
}

/*
 * @brief   通过网络写日志的具体实现
 */
int log_client_write_impl(int fd, enum log_level lv, const char * content)
{
    struct log_head head;

    if (0 != _send_write_cmd(fd, lv, content)) {
        printf("Send write cmd failed.\n");
        return -1;
    }

    memset(&head, 0, sizeof(head));
    if (0 <= read(fd, &head, LOG_HEAD_SIZE))
        return -1;

    printf("Answer for write: %s %s %s\n", head.version, head.command,
            head.length);

    return 0;
}

static inline int _send_download_cmd(int fd)
{
    struct log_head head;
    memset(&head, 0, sizeof(head));

    sprintf(head.version, "%s", LOG_VERSION);
    sprintf(head.command, "%s", "DOWNLOAD");
    sprintf(head.length, "%d", 0);

    printf("Send download cmd: %s %s %s\n", head.version, head.command,
            head.length);

    return _write_back(fd, (char *) &head, LOG_HEAD_SIZE);
}

/*
 * @brief   下载日志文件到本地fname指定的文件中
 *
 * @return  0,成功且写入完成
 *          -1: fd错误，且fd应当被关闭不再使用
 *          -2: fname相关错误，如文件打开失败等。
 */
int log_client_download(int fd, const char *fname)
{
    struct log_head head;
    int ret;
    int len;
    char temp[1024];
    int count = 0;
    int flag = 0;

    if (0 != _send_download_cmd(fd)) {
        printf("Write to server failed(download).\n");
        return -1;
    }

    memset(&head, 0, sizeof(head));
    ret = read(fd, &head, LOG_HEAD_SIZE);
    if (ret != LOG_HEAD_SIZE)
        return -1;

    printf("Answer for download: %s %s %s %s\n", head.version, head.command,
            head.length, &(head.content_start));
    flag = strcmp(head.command, "FILE") == 0;
    len = atoi(head.length);

    if (len <= 0)
        return 0;

    FILE * pf = fopen(fname, "wb");

    while (count < len) {
        memset(temp, 0, sizeof(temp));
        ret = read(fd, temp, 1024);
        if (ret <= 0)
            break;
        if (flag && pf != NULL) {
            fwrite(temp, ret, 1, pf);
        } else {
            printf("%s\n", temp);
        }
        count += ret;
    }

    if (pf != NULL) {
        fclose(pf);
        return 0;
    } else {
        return -2;
    }
}

/*
 * @brief   以无连接的形式发送日志，不需要先创建客户端
 */
int log_client_write2_impl(const char *server, enum log_level lv,
        const char * content)
{
    static int fd = -1;
    struct sockaddr_in addr;
    struct log_head *phead;
    char buf[MAX_LOG_CMD_LEN + MAX_LOG_LEN];
    int len;

    if (server == NULL || content == NULL || strlen(content) == 0)
        return -1;

    if (fd == -1)
        fd = socket(AF_INET, SOCK_DGRAM, 0);

    if (-1 == fd)
        return -1;

    addr.sin_family = AF_INET;
    addr.sin_port = htons(LOG_PORT);
    addr.sin_addr.s_addr = inet_addr(server);

    len = LOG_HEAD_SIZE + strlen(content);
    phead = (struct log_head *) buf;
    sprintf(phead->version, LOG_VERSION);
    sprintf(phead->command, "LOG %d", lv);
    sprintf(&(phead->content_start), "%s", content);
    sprintf(phead->length, "%d", (int) strlen(content));

    if (0 > sendto(fd, buf, len, 0, (struct sockaddr*) &addr, sizeof(addr))) {
        close(fd);
        fd = -1;
    }

    return 0;
}

/*
 * @brief   释放查询函数中返回的指针
 */
void log_client_freebuf(char *buf)
{
    if (buf != NULL)
        free(buf);
}

#endif

#ifdef LOG_SERVER_CODE

#define MAX_QUERY_FDS   256
#define ERROR_FD    -1

/*
 * @brief   全局存放已连接TCP客户端的fd，由此可见这个数组的大小决定了可同时访问客户端的最大个数
 */
static int g_query_fds[MAX_QUERY_FDS]; // 保证最大的fd放在第一个(idx = 0)
static pthread_mutex_t g_lock = PTHREAD_MUTEX_INITIALIZER;

#define LOCK_FDS    pthread_mutex_lock(&g_lock);
#define UNLOCK_FDS    pthread_mutex_unlock(&g_lock);

/*
 * @brief   将收到的TCP连接fd放入全局数组中
 * @return  成功返回0，失败返回-1
 */
static inline int _put_fd(int peer_fd)
{
    int i;
    LOCK_FDS
    for (i = 0; i < MAX_QUERY_FDS; i++) {
        if (g_query_fds[i] != ERROR_FD)
            continue;

        // 此处保证最大的fd放在第一个(idx = 0)
        if (g_query_fds[0] < peer_fd) {
            g_query_fds[i] = g_query_fds[0];
            g_query_fds[0] = peer_fd;
        } else {
            g_query_fds[i] = peer_fd;
        }

        UNLOCK_FDS
        return 0;
    }
    UNLOCK_FDS
    return -1;
}

static inline void _remove_fd(int peer_fd)
{
    int i;
    LOCK_FDS
    for (i = 0; i < MAX_QUERY_FDS; i++) {
        if (g_query_fds[i] == peer_fd) {
            g_query_fds[i] = ERROR_FD;
            break;
        }
    }
    UNLOCK_FDS
}

/*
 * @brief   socket\bind\listen三合一
 */
static inline int _create_listen_server()
{
    int sock = -1;
    struct sockaddr_in addr = { AF_INET, htons(LOG_PORT), { INADDR_ANY } };

    sock = socket(AF_INET, SOCK_STREAM, 0);
    if (sock == -1) {
        log_impl(LOG_FATAL, "LogServer: create socket failed, thread exit.");
        return -1;
    }

    int one = 1;
    setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));

    if (-1 == bind(sock, (struct sockaddr *) &addr, sizeof(addr))) {
        log_impl(LOG_FATAL, "LogServer: bind(9528) failed, thread exit.");
        close(sock);
        return -1;
    }

    if (-1 == listen(sock, 5)) {
        log_impl(LOG_FATAL, "LogServer: listen failed, thread exit.");
        close(sock);
        return -1;
    }

    return sock;
}

/*
 * @brief   判断字符串str的开头是否是字符串prefix
 * @return  是返回1 否返回0
 */
static inline int _str_startwith(const char * str, const char * prefix)
{
    int i;

    for (i = 0; str[i] != 0 && prefix[i] != 0; i++) {
        if (str[i] != prefix[i])
            return 0;
    }

    return 1;
}

static inline void _send_result(int fd, const char * cmd, const char * result)
{
    struct log_head head;
    memset(&head, 0, sizeof(head));

    sprintf(head.version, "%s", LOG_VERSION);
    sprintf(head.command, "%s", cmd);
    sprintf(head.length, "%d", result == NULL ? 0 : (int) strlen(result));

    if (0 != _write_back(fd, (char *) &head, LOG_HEAD_SIZE)) {
        log_error("LogServer: send head to client failed.");
        close(fd);
        return;
    }

    if (result == NULL)
        return;

    if (0 != _write_back(fd, result, strlen(result))) {
        log_error("LogServer: send content to client failed.");
        close(fd);
        return;
    }
}

/*
 * @brief   创建TCP服务端，并接收新到的连接，将连接的客户端fd保存至全局的缓存g_query_fds中
 */
static void * _tcp_thread(void * arg)
{
    int server_fd = _create_listen_server();
    if (server_fd == -1)
        return NULL;

    while (1) {
        int peerfd;
        struct sockaddr peeraddr;
        socklen_t addrlen;

        peerfd = accept(server_fd, &peeraddr, &addrlen);

        if (0 != _put_fd(peerfd)) {
            const char * err_str = "server busy, too many connections";
            _send_result(peerfd, "ERROR", err_str);
            close(peerfd); // 无法放入,直接关闭
            log_error("LogServer: server busy, close new client.");
        }
    }
    return NULL;
}

static void _query(int fd, int len)
{
    enum log_level lv_begin = 0;
    enum log_level lv_end = 0;
    time_t time_begin = 0;
    time_t time_end = 0;
    int max = 0;
    int ret;
    char cmd[MAX_LOG_CMD_LEN];

    memset(cmd, 0, sizeof(cmd));

    assert(len < MAX_LOG_CMD_LEN);

    if (0 != _read_n_bytes(fd, cmd, len)) {
        log_error("LogServer: query, read client failed(len = %d)", len);
        close(fd);
        return;
    }

    ret = sscanf(cmd, "%d %d %ld %ld %d", (int*) (&lv_begin), (int *) (&lv_end),
            &time_begin, &time_end, &max);

    if (5 != ret) {
        static const char * err_msg = "ERROR unknow query format.";
        _send_result(fd, "ERROR", err_msg);
        return;
    }

    printf("QUERY %d %d %ld %ld %d\n", lv_begin, lv_end, time_begin, time_end,
            max);

    char *buf = NULL;
    ret = log_query(lv_begin, lv_end, time_begin, time_end, max, &buf);
    _send_result(fd, "RESULT", buf);
    if (buf != NULL)
        log_freebuf(buf);
}

static void _stat_query(int fd)
{
    char buf[MAX_LOG_CMD_LEN];
    int count = 0;
    time_t time_begin = 0;
    time_t time_end = 0;

    log_query_stat(&count, &time_begin, &time_end);

    sprintf(buf, "%d %ld %ld", count, time_begin, time_end);

    _send_result(fd, "STATE", buf);
}

static void _log(int fd, enum log_level lv, int len)
{
    char temp[MAX_LOG_LEN];
    memset(temp, 0, sizeof(temp));

    _send_result(fd, "OK", NULL);

    if (len <= 0)
        return;

    if (0 != _read_n_bytes(fd, temp, len)) {
        log_error("LogServer: log, read from client failed(len = %d)", len);
        close(fd);
        return;
    }

    log_impl(lv, temp);
}

static inline long cm_flength(FILE *file)
{
    long pos = ftell(file);
    fseek(file, 0, SEEK_END);
    long len = ftell(file);
    fseek(file, pos, SEEK_SET);

    return len;
}

static void _download(int fd)
{
    FILE *pf;
    char buf[4096];
    long len;
    int ret;
    int n;

    //下载前关闭数据库，此时写日志等调用均直接返回
    LOCK_DB;
    if (g_database != NULL)
        sqlite3_close(g_database);
    g_database = NULL;
    UNLOCK_DB;

    pf = fopen(g_log_file, "rb");
    if (pf == NULL) {
        _send_result(fd, "ERROR", "Open file failed.");
        goto _L_REOPEN;
    }

    len = cm_flength(pf);
    if (len <= 0) {
        _send_result(fd, "ERROR", "file is empty.");
        fclose(pf);
        goto _L_REOPEN;
    }

    struct log_head head;
    memset(&head, 0, sizeof(head));
    sprintf(head.version, "%s", LOG_VERSION);
    sprintf(head.command, "FILE");
    sprintf(head.length, "%ld", len);
    if (0 != _write_back(fd, (char *) &head, LOG_HEAD_SIZE)) {
        close(fd);
        goto _L_REOPEN;
    }

    while (len > 0) {
        ret = fread(buf, 1, 4096, pf);
        if (ret > 0) {
            if (0 != _write_back(fd, buf, ret)) {
                close(fd);
                goto _L_REOPEN;
            }
            len -= ret;
        } else {
            memset(buf, 0, sizeof(buf));
            n = len > 4096 ? 4096 : len;
            if (0 != _write_back(fd, buf, n)) {
                close(fd);
                goto _L_REOPEN;
            }
            len -= n;
        }
    }

    _L_REOPEN: log_setup(NULL);
}

/*
 * @brief   响应有数据到来的TCO连接
 */
static inline void _tcp_response(int client_fd)
{
    struct log_head head;
    int ret;
    char temp[MAX_LOG_CMD_LEN];
    enum log_level lv;
    static const char * err_msg = "unknow command (or format).";

    memset(&head, 0, sizeof(head));
    ret = read(client_fd, &head, LOG_HEAD_SIZE);

    if (ret != LOG_HEAD_SIZE) {
        close(client_fd);
        _remove_fd(client_fd);
//        printf("Read header failed, close connection.\n");
//        printf("head: %s %s %s %s\n", head.version, head.command, head.length,
//                (&head.content_start));
        log_info("Read header failed, close connection.");
        return;
    }

//    printf("recvfrom client: command = %s\n", head.command);

    if (strcmp(head.command, "QUERY") == 0) {
        _query(client_fd, atoi(head.length));
    } else if (strcmp(head.command, "STAT_QUERY") == 0) {
        _stat_query(client_fd);
    } else if (_str_startwith(head.command, "LOG")) {
        if (2 != sscanf(head.command, "%s %d", temp, (int *) (&lv)))
            _send_result(client_fd, "ERROR", err_msg);
        else
            _log(client_fd, lv, atoi(head.length));
    } else if (strcmp(head.command, "DOWNLOAD") == 0) {
        _download(client_fd);
    } else {
        _send_result(client_fd, "ERROR", err_msg);
    }
}

/*
 * @brief   辅助函数，将全局缓存的fd设置到set中，并返回最大的fd
 *          并且在fds中返回一个紧凑排列的集合（全局缓存的有空洞）
 */
static inline int _mk_fdset(fd_set * set, int *max_fd, int fds[MAX_QUERY_FDS])
{
    int i, j;
    int copyed_fds[MAX_QUERY_FDS];

    FD_ZERO(set);

    LOCK_FDS
    memcpy(copyed_fds, g_query_fds, sizeof(copyed_fds));
    UNLOCK_FDS

    *max_fd = copyed_fds[0];
    if (*max_fd == ERROR_FD)
        return -1;

    for (i = 0, j = 0; i < MAX_QUERY_FDS; i++) {
        if (copyed_fds[i] != ERROR_FD) {
            FD_SET(copyed_fds[i], set);
            fds[j++] = copyed_fds[i];
        }
    }

    fds[j] = ERROR_FD; // 结束标识

    return 0;
}

static void * _tcp_response_thread()
{
    fd_set readset;
    int max_fd;
    int fds[MAX_QUERY_FDS];
    struct timeval timeout;

    int i;
    int ret;

    while (1) {
        timeout.tv_sec = 2;
        timeout.tv_usec = 0;
        if (0 != _mk_fdset(&readset, &max_fd, fds)) {
            usleep(200 * 1000);
            continue;
        }

        ret = select(max_fd + 1, &readset, NULL, NULL, &timeout);
        if (ret <= 0)
            continue;

        for (i = 0; fds[i] != ERROR_FD; i++) {
            if (FD_ISSET(fds[i], &readset)) {
                printf("read [%d] ok\n", fds[i]);
                _tcp_response(fds[i]);
            }
        }

    }
    return NULL;
}

/*
 * @brief   从UDP接收写命令，一次写入一条日志
 */
static void * _udp_thread(void *arg)
{
    struct sockaddr_in addr = { AF_INET, htons(LOG_PORT), { INADDR_ANY } };
    char buf[MAX_LOG_CMD_LEN + MAX_LOG_LEN];
    char cmd[MAX_LOG_CMD_LEN];
    enum log_level lv;
    struct log_head * phead;
    int len;
    int one = 1;

    int fd = socket(AF_INET, SOCK_DGRAM, 0);
    if (fd == -1) {
        log_error("LogServer: create udp socket failed.");
        return NULL;
    }

    setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));

    if (-1 == bind(fd, (struct sockaddr *) &addr, sizeof(addr))) {
        log_error("LogServer: bind (udp:%d) failed.", LOG_PORT);
        return NULL;
    }

    while (1) {
        memset(buf, 0, sizeof(buf));
        len = recvfrom(fd, buf, sizeof(buf), 0, NULL, NULL);
//        printf("udp log(%d): %s\n", len, buf);
        if (len < LOG_HEAD_SIZE) {
            log_error("LogServer: udp data len error.");
            continue;
        }

        phead = (struct log_head *) buf;
//        printf("%s %s %s\n", phead->version, phead->command,
//                &phead->content_start);
        if (2 != sscanf(phead->command, "%s %d", cmd, (int *) (&lv))) {
            log_error("LogServer: udp data format error.");
            continue;
        }

        if (atoi(phead->length) <= 0) {
            log_error("LogServer: udp data with empty content.");
            continue;
        }

        log_impl(lv, &phead->content_start);
//        printf("udp log end....\n");
    }
    close(fd);
    return NULL;
}

/*
 * @brief   启动服务端
 * @warning 多次调用只启动一个服务器实例
 */
void log_server_start()
{
    static int already_started = 0;
    if (already_started)
        return;

    int i;
    for (i = 0; i < MAX_QUERY_FDS; i++)
        g_query_fds[i] = ERROR_FD;

    pthread_t tid;
    pthread_attr_t attr;
    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);

    pthread_create(&tid, &attr, _tcp_thread, NULL);
    pthread_create(&tid, &attr, _tcp_response_thread, NULL);
    pthread_create(&tid, &attr, _udp_thread, NULL);

    already_started = 1;

    pthread_attr_destroy(&attr);
}

#endif

